package cn.spark.study.core;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

/**
 * 分组取top3
 *
 * @author jun.zhang6
 * @date 2020/11/6
 */
public class GroupTop3 {

    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf()
                .setAppName("GroupTop3")
                .setMaster("local");

        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        JavaRDD<String> linesRdd = sc.textFile("C:\\Users\\jun.zhang6\\Desktop\\score.txt");

        JavaPairRDD<String, Integer> pairsRdd = linesRdd.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String line) throws Exception {
                String[] split = line.split(" ");
                return new Tuple2<String, Integer>(split[0], Integer.valueOf(split[1]));
            }
        });

        JavaPairRDD<String, Iterable<Integer>> groupedRdd = pairsRdd.groupByKey();

        JavaPairRDD<String, Iterable<Integer>> top3ScoreRdd = groupedRdd.mapToPair(new PairFunction<Tuple2<String, Iterable<Integer>>, String, Iterable<Integer>>() {
            @Override
            public Tuple2<String, Iterable<Integer>> call(Tuple2<String, Iterable<Integer>> tuple) throws Exception {
                String className = tuple._1();
                Iterator<Integer> scores = tuple._2().iterator();
                Integer[] top3 = new Integer[3];

                while (scores.hasNext()) {
                    Integer score = scores.next();

                    for (int i = 0; i < top3.length; i++) {
                        if (top3[i] == null) {
                            top3[i] = score;
                            break;
                        } else if (score > top3[i]) {
                            for (int j = 2; j > i; j--) {
                                top3[j] = top3[j - 1];
                            }
                            top3[i] = score;
                            break;
                        }
                    }
                }
                return new Tuple2<String, Iterable<Integer>>(className, Arrays.asList(top3));
            }
        });

        top3ScoreRdd.foreach(new VoidFunction<Tuple2<String, Iterable<Integer>>>() {
            @Override
            public void call(Tuple2<String, Iterable<Integer>> tuple) throws Exception {
                System.out.println("class: " + tuple._1());
                Iterator<Integer> iterator = tuple._2().iterator();
                while (iterator.hasNext()) {
                    Integer score = iterator.next();
                    System.out.println(score);
                }
                System.out.println("=======================================");
            }
        });

        sc.close();
    }
}
